package com.lhc.kafkademo.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import scala.collection.immutable.Stream;

import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;


public class StreamSample {

    /**
     * 先创建topic
     */
    private static final String INPUT_TOPIC = "stream-in";
    private static final String OUT_TOPIC = "stream-out";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.11.130:9092");
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"wordcount-app");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 如何构建流结构拓扑
        StreamsBuilder builder = new StreamsBuilder();

        //构建wordcount process
        wordCountStream(builder);

        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties);
        kafkaStreams.start();

    }

    //如何定义流计算过程
    static void wordCountStream(final StreamsBuilder builder){
        // 不断从INPUT_TOPIC上获取新数据，并且追加到流上一个抽象对象
        KStream<String, String> source = builder.stream(INPUT_TOPIC);
        //KTable 是数据集合的抽象对象
        KTable<String, Long> count =
                source
                        .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" ")))
                        .groupBy((key, value) -> value)
                        .count();
        count.toStream().to(OUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
    }

    /**
     * shell 测试脚本
     * kafka-console-producer.sh --broker-list 192.168.11.130:9092 --topic stream-in
     * kafka-console-consumer.sh --bootstrap-server 192.168.11.130:9092 \
     *  --topic stream-out \
     *  --property print.key=true \
     *  --property print.value=true \
     *  --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
     *  --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer \
     *  --from-beginning
     */

}
